package com.wyw.hemerocallis.rocket;

import com.alibaba.fastjson.JSON;
import com.wyw.hemerocallis.vo.WywMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * 通过RocketMQ发送消息
 * 基于Spring Messaging
 *
 * @author Mr Wu    yewen.wu.china@gmail.com
 * <p>
 * Update History:
 * Author        Time            Content
 */
@Slf4j
@Component
public class RocketMQProducer {
    /**
     * 类似Kafka中的topic，默认的读写队列都是四个
     */
    private static final String TOPIC = "wyw-study-rocketmq";

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 使用同步的方式发送消息，不指定key和tag
     */
    public void sendMessageWithValue(String value) {
        //随机选择一个Topic的Message Queue发送消息
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC, value);
        log.info("sendMessageWithValue result: [{}]", JSON.toJSONString(sendResult));

        SendResult sendResultOrderly = rocketMQTemplate.syncSendOrderly(
                TOPIC, value, "wyw"
        );
        log.info("sendMessageWithValue orderly result: [{}]",
                JSON.toJSONString(sendResultOrderly));
    }

    /**
     * 使用异步的方式发送消息，指定key(索引键)
     * hashKey分区效果
     * "a b"    key，前缀，多个用空格分隔
     * 可见源码RocketMQUtil
     */
    public void sendMessageWithKey(String key, String value) {
        Message<String> message = MessageBuilder.withPayload(value)
                .setHeader(RocketMQHeaders.KEYS, key).build();

        //异步发送消息，并设置回调
        rocketMQTemplate.asyncSend(TOPIC, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("sendMessageWithKey success result: [{}]",
                        JSON.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable e) {
                log.error("sendMessageWithKey failure: [{}]", e.getMessage(), e);
            }
        });

    }

    /**
     * 使用同步的方式发送消息，带有tag，且发送的是Java Pojo
     */
    public void sendMessageWithTag(String tag, String value) {
        WywMessage wywMessage = JSON.parseObject(value, WywMessage.class);

        SendResult sendResult = rocketMQTemplate.syncSend(
                String.format("%s:%s", TOPIC, tag), // tag消费者组的概念，kafka不存在，实际使用场景不常见
                wywMessage
        );
        log.info("sendMessageWithTag result: [{}]", JSON.toJSONString(sendResult));
    }

    /**
     * 使用同步的方式发送消息, 带有 key 和 tag
     * */
    public void sendMessageWithAll(String key, String tag, String value) {

        Message<String> message = MessageBuilder.withPayload(value)
                .setHeader(RocketMQHeaders.KEYS, key).build();
        SendResult sendResult = rocketMQTemplate.syncSend(
                String.format("%s:%s", TOPIC, tag),
                message
        );
        log.info("sendMessageWithAll result: [{}]", JSON.toJSONString(sendResult));
    }

}
